# uncompyle6 version 3.2.3
# Python bytecode 3.6 (3379)
# Decompiled from: Python 3.6.8 |Anaconda custom (64-bit)| (default, Feb 21 2019, 18:30:04) [MSC v.1916 64 bit (AMD64)]
# Embedded file name: api\handlers\frontend\scanner.py
import logging, uuid, datetime
from dateutil.tz import tzlocal
from sqlalchemy import select as sql_select
from sqlalchemy import asc as sql_asc
from sqlalchemy import or_ as sql_or
import sqlalchemy.exc
from api.handlers.frontend import FrontendResourceHandler
from db import Connection
from db.tables.profiles import ProfileRow, ProfilesTable
from api.classes.frontend.auto import *
from db.data.users import get_owner_id, require_at_least_tech_admin_with_access_all
from server.http_responses.http_status_codes import Http201, Http204
from server.http_responses.http_errors import BadApiRequest, Http404, Http409
from helpers.licensing.features import Features
from helpers.constants.jobs import ScanAppTypes

logger = logging.getLogger("frontend_api_resources.Scanner")
known_scanning_apps = (ScanAppTypes.WVS, ScanAppTypes.OVAS)
get_profiles_query = (
    sql_select(
        (ProfileRow.profile_id, ProfileRow.name, ProfileRow.jobs, ProfileRow.sort_order)
    )
    .where(ProfileRow.sort_order > 0)
    .where(ProfileRow.deleted_at.is_(None))
    .order_by(sql_asc(ProfileRow.sort_order))
)


class Scanner(FrontendResourceHandler):
    def _get_scanning_profiles(self, scanning_profile_id=None):
        owner_id = get_owner_id(self.request_user)
        query = get_profiles_query.where(
            sql_or(ProfileRow.owner_id == owner_id, ProfileRow.owner_id.is_(None))
        )
        if scanning_profile_id:
            query = query.where(ProfileRow.profile_id == scanning_profile_id)
        result = []
        with Connection(self.current_shard_name) as (db):
            for profile in db.execute(query).fetchall():
                profile = dict(profile)
                profile["custom"] = not profile["profile_id"].startswith(
                    "11111111-1111-1111-1111-"
                )
                jobs = profile["jobs"]
                if jobs is None:
                    jobs = {}
                checks = []
                if profile["custom"]:
                    for scanning_app in known_scanning_apps:
                        if scanning_app in jobs:
                            for check in jobs[scanning_app]["checks"]:
                                if check.startswith("+"):
                                    check = "+%s%s" % (scanning_app, check[1:])
                                else:
                                    check = scanning_app + check
                                checks.append(check)

                        else:
                            checks.append(scanning_app + "/")

                profile["checks"] = checks
                result.append(ClsScanningProfile(**profile))

        return result

    def get_scanning_profiles(self, **_):
        return ClsScanningProfilesResponse(
            scanning_profiles=self._get_scanning_profiles()
        )

    def get_scanning_profile(self, scanning_profile_id, **_):
        profiles = self._get_scanning_profiles(scanning_profile_id=scanning_profile_id)
        if not profiles:
            raise Http404()
        return profiles[0]

    def delete_scanning_profile(self, scanning_profile_id, **_):
        if scanning_profile_id.startswith("11111111"):
            raise Http409()
        owner_id = get_owner_id(self.request_user)
        self.license.require(Features.SCANNING_PROFILES)
        require_at_least_tech_admin_with_access_all(self.request_user)
        query = get_profiles_query.where(
            sql_or(ProfileRow.owner_id == owner_id, ProfileRow.owner_id.is_(None))
        )
        query = query.where(ProfileRow.profile_id == scanning_profile_id)
        query = query.where(ProfileRow.sort_order > 0)
        with Connection(self.current_shard_name) as (db):
            profile = db.execute(query).fetchone()
            if not profile:
                raise Http404()
            query = (
                ProfilesTable.update(
                    values=dict(deleted_at=datetime.datetime.now(tzlocal()))
                )
            ).where(ProfileRow.profile_id == scanning_profile_id)
            db.execute(query)
        return Http204()

    @staticmethod
    def _process_checks(checks):
        try:
            jobs = dict()
            for check in checks:
                if check.startswith("#"):
                    continue
                if check.startswith("+"):
                    scanning_app, check = check[1:].split("/", 1)
                    check = "+/" + check
                else:
                    scanning_app, check = check.split("/", 1)
                    check = "/" + check
                if scanning_app not in known_scanning_apps:
                    raise BadApiRequest("Unknown check")
                if scanning_app not in jobs:
                    jobs[scanning_app] = dict(profile="custom", checks=[])
                jobs[scanning_app]["checks"].append(check)

            to_remove_scanning_apps = []
            for scanning_app in known_scanning_apps:
                if scanning_app not in jobs:
                    jobs[scanning_app] = dict(profile="custom", checks=[])
                elif "/" in jobs[scanning_app]["checks"]:
                    for check in jobs[scanning_app]["checks"]:
                        if check.startswith("+"):
                            break
                    else:
                        to_remove_scanning_apps.append(scanning_app)

            for scanning_app in to_remove_scanning_apps:
                del jobs[scanning_app]

            try:
                del jobs[ScanAppTypes.OVAS]
            except KeyError:
                pass

            if not jobs:
                raise BadApiRequest("no valid check found")
            return jobs
        except BadApiRequest:
            raise
        except Exception as e:
            logger.error("invalid checks provided %s", e)
            raise BadApiRequest("Invalid checks")

    def create_scanning_profile(self, body, **_):
        owner_id = get_owner_id(self.request_user)
        self.license.require(Features.SCANNING_PROFILES)
        require_at_least_tech_admin_with_access_all(self.request_user)
        profile = ClsScanningProfile(**body)
        jobs = self._process_checks(profile.get("checks", list()))
        profile_data = dict(
            profile_id=str(uuid.uuid4()),
            owner_id=owner_id,
            creator_id=self.request_user["user_id"],
            name=profile.name,
            sort_order=profile.sort_order,
            jobs=jobs,
        )
        q = ProfilesTable.insert().values(**profile_data)
        with Connection(self.current_shard_name) as (db):
            try:
                db.execute(q)
            except sqlalchemy.exc.IntegrityError as e:
                if "uix_scanning_profiles" in str(e):
                    raise BadApiRequest("duplicate profile name")

        profile.custom = True
        profile.profile_id = profile_data["profile_id"]
        return Http201(
            location=self.base_path
            + "/scanning_profiles/"
            + profile_data["profile_id"],
            body=profile,
        )

    def update_scanning_profile(self, scanning_profile_id, body, **_):
        if scanning_profile_id.startswith("11111111"):
            raise Http409()
        self.license.require(Features.SCANNING_PROFILES)
        owner_id = get_owner_id(self.request_user)
        require_at_least_tech_admin_with_access_all(self.request_user)
        query = get_profiles_query.where(
            sql_or(ProfileRow.owner_id == owner_id, ProfileRow.owner_id.is_(None))
        )
        query = query.where(ProfileRow.profile_id == scanning_profile_id)
        query = query.where(ProfileRow.sort_order > 0)
        profile = ClsScanningProfile(**body)
        jobs = self._process_checks(profile.get("checks", list()))
        with Connection(self.current_shard_name) as (db):
            old_profile = db.execute(query).fetchone()
            if not old_profile:
                raise Http404()
            update = dict()
            if old_profile.name != profile.name:
                update["name"] = profile.name
            if old_profile.sort_order != profile.sort_order:
                update["sort_order"] = profile.sort_order
            if old_profile.jobs != jobs:
                update["jobs"] = jobs
            query = update and (ProfilesTable.update(values=update)).where(
                ProfileRow.profile_id == scanning_profile_id
            )
            try:
                db.execute(query)
            except sqlalchemy.exc.IntegrityError as e:
                if "uix_scanning_profiles" in str(e):
                    raise BadApiRequest("duplicate profile name")

        return Http204()
